package com.chenqq.qsocket.mina;

import android.text.TextUtils;
import android.util.Log;

import com.chenqq.qsocket.adapter.TcpBasicAdapter;
import com.chenqq.qsocket.configuration.MinaConfigInfo;
import com.chenqq.qsocket.fractory.HeartBeatKeepAliveFilter;
import com.chenqq.qsocket.listener.SocketConnectListener;
import com.chenqq.qsocket.listener.SocketMessageListener;
import com.chenqq.qsocket.message.RequestSendFileMessage;
import com.chenqq.qsocket.tools.FileSender;
import com.chenqq.qsocket.tools.LogShow;

import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.service.IoServiceListener;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.keepalive.KeepAliveFilter;
import org.apache.mina.filter.keepalive.KeepAliveMessageFactory;
import org.apache.mina.filter.keepalive.KeepAliveRequestTimeoutHandler;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import java.io.File;
import java.net.InetSocketAddress;

import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.BehaviorSubject;

/**
 * TCP长连接客户端
 */
public class MinaSocketClient {
    private NioSocketConnector mSocketConnector;//连接客户端
    private String tag = MinaSocketClient.this.getClass().getSimpleName();
    private int Port = 5566;
    private String tcpHostAddress = "";//目标地址
    private ConnectFuture mFuture;
    private BehaviorSubject<Object> tcpBehaviorSubjectClient;
    private SocketMessageListener socketSendListener;
    private SocketConnectListener connectListener;
    private BehaviorSubject<Object> messageBehaviorSubject;

    public MinaSocketClient() {
        mSocketConnector = new NioSocketConnector();
    }

    public MinaSocketClient setHandler(IoHandlerAdapter ioHandlerAdapter) {
        mSocketConnector.setHandler(ioHandlerAdapter);
        return this;
    }

    public MinaSocketClient setSocketSendListener(SocketMessageListener socketSendListener) {
        this.socketSendListener = socketSendListener;
        return this;
    }

    public MinaSocketClient setConnectListener(SocketConnectListener connectListener) {
        this.connectListener = connectListener;
        return this;
    }

    public MinaSocketClient setTag(String tag) {
        this.tag = tag;
        return this;
    }
    public MinaSocketClient setTcpNoDelay(boolean isDelay){
        mSocketConnector.getSessionConfig().setTcpNoDelay(isDelay);
        return this;
    }

    public MinaSocketClient setFilterChain(String name, IoFilter filter) {
        mSocketConnector.getFilterChain().addLast(name, filter);
        return this;
    }

    public MinaSocketClient setKeepAliveFilter(int intervalTime, int timeOut
            , KeepAliveMessageFactory keepAliveMessageFactory) {
        HeartBeatKeepAliveFilter heartFilter =
                new HeartBeatKeepAliveFilter(keepAliveMessageFactory,IdleStatus.READER_IDLE
        , KeepAliveRequestTimeoutHandler.CLOSE);
        //每 5 分钟发送一个心跳包
        heartFilter.setRequestInterval(intervalTime);
        //心跳包超时时间 10s
        heartFilter.setForwardEvent(false);
        heartFilter.setRequestTimeout(timeOut);
        mSocketConnector.getSessionConfig().setKeepAlive(true);
        mSocketConnector.getFilterChain().addLast("heartbeat", heartFilter);
        return this;
    }

    public MinaSocketClient setIoServiceListener(IoServiceListener ioServiceListener) {
        mSocketConnector.addListener(ioServiceListener);
        return this;
    }
    public MinaSocketClient setIdleTime(IdleStatus idleStatus, int time) {
        mSocketConnector.getSessionConfig().setIdleTime(idleStatus, time);
        return this;
    }

    public MinaSocketClient setPort(int port) {
        this.Port = port;
        return this;
    }

    public MinaSocketClient setTcpHostAddress(String tcpHostAddress) {
        this.tcpHostAddress = tcpHostAddress;
        return this;
    }

    public void start() {
        if (tcpBehaviorSubjectClient == null) {
            tcpBehaviorSubjectClient = BehaviorSubject.create();
            Disposable subscribe = tcpBehaviorSubjectClient.unsubscribeOn(Schedulers.io())
                    .subscribeOn(Schedulers.io())
                    .map(new Function<Object, String>() {
                        @Override
                        public String apply(@NonNull Object o) throws Exception {
                            if (mSocketConnector == null) {
                                return "数据端已经清除";
                            }
                            if (TextUtils.isEmpty(tcpHostAddress)) {
                                return "目标IP地址为空";
                            }
                            IoHandler handler = mSocketConnector.getHandler();
                            if (handler instanceof TcpBasicAdapter) {
                                ((TcpBasicAdapter) handler).setSocketMessageListener(socketSendListener)
                                        .setTag(tag).setFileDown(MinaConfigInfo.getInStance().getTemporaryPath()
                                        , false)
                                        .setFileDown(MinaConfigInfo.getInStance().getPathDown(), true);
                            }
                            if (mFuture == null || mFuture.isCanceled()
                                    || !mFuture.isConnected()) {
                                InetSocketAddress mSocketAddress =
                                        new InetSocketAddress(tcpHostAddress, Port);
                                mFuture = mSocketConnector.connect(mSocketAddress);
                                mFuture.awaitUninterruptibly();
                            }

                            if (mFuture.isConnected() && mFuture.isDone()) {
                                return "Success:服务器连接成功" + tcpHostAddress;
                            }
                            return "TCP连接失败：" + tcpHostAddress + "  :" + Port;
                        }
                    }).observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(String s) throws Exception {
                            Log.e("TCP", s);
                            if (s.startsWith("Success")) {
                                if (connectListener != null) {
                                    connectListener.isConnect();
                                }
                            } else {
                                if (connectListener != null) {
                                    connectListener.connectFailed(tag, s);
                                }
                            }

                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            Log.e("TCP错误", throwable.getMessage());
                            if (connectListener != null) {
                                connectListener.connectFailed(tag, throwable.toString());
                            }
                            MinaConfigInfo.getInStance().removeDisposable(tag);
                            tcpBehaviorSubjectClient = null;
                        }
                    });
            MinaConfigInfo.getInStance().addDisposable(tag, subscribe);
        }
        tcpBehaviorSubjectClient.onNext(true);
    }

    public void sendMessage(Object message) {
        if (message == null) {
            return;
        }
        if (message instanceof String) {
            writeMessage(message);
        } else if (message instanceof File) {
            File file = (File) message;
            sendFile(file.getAbsolutePath());
        }
    }


    public void sendFile(String path) {
        FileSender fileSender = new FileSender();
        fileSender.sendFile(path, new FileSender.SendFileListener() {
            @Override
            public void onSucess(RequestSendFileMessage requestSendFileMessage) {
                writeMessage(requestSendFileMessage);
            }

            @Override
            public void onFailed(String hint) {
                if (socketSendListener!=null){
                    socketSendListener.sendMessageFail(tag,hint);
                }
            }
        });
    }

    private  void writeMessage(Object o){
        if (messageBehaviorSubject==null) {
            messageBehaviorSubject = BehaviorSubject.create();
            Disposable subscribe = messageBehaviorSubject.unsubscribeOn(Schedulers.io())
                    .observeOn(Schedulers.io())
                    .subscribeOn(Schedulers.io()).map(new Function<Object, Object>() {
                @Override
                public Object apply(@NonNull Object o) throws Exception {
                    if (tcpBehaviorSubjectClient == null ||mFuture==null
                            ||mFuture.isCanceled() || !mFuture.isConnected()
                    ||!mFuture.getSession().isConnected()) {
                        InetSocketAddress mSocketAddress =
                                new InetSocketAddress(tcpHostAddress, Port);
                        mFuture = mSocketConnector.connect(mSocketAddress);
                        mFuture.awaitUninterruptibly();
                    }
                    if (mFuture.isConnected()&&mFuture.isDone()) {

                        mFuture.getSession().write(o);
                        return true;
                    }
                    return "服务器未连接";
                }
            }).subscribe(new Consumer<Object>() {
                @Override
                public void accept(Object o) throws Exception {
                    if (o instanceof String){
                        LogShow.showE("=============="+o);
                        if (socketSendListener!=null){
                            socketSendListener.sendMessageFail(tag,(String) o);
                        }
                    }else {
                        LogShow.showV("=============="+o);
                    }

                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    if (socketSendListener!=null){
                        socketSendListener.sendMessageFail(tag,(String) throwable.getMessage());
                    }
                    LogShow.showE("=============="+throwable.getMessage());
                    MinaConfigInfo.getInStance().removeDisposable("writeMessage");
                    messageBehaviorSubject=null;
                }
            });
            MinaConfigInfo.getInStance().addDisposable("writeMessage",subscribe);
        }
        messageBehaviorSubject.onNext(o);
    }


    public void stop() {
        if (mFuture != null) {
            mFuture.cancel();
            mFuture = null;
        }
        if (mSocketConnector != null) {
            mSocketConnector.dispose();
            mSocketConnector = null;
        }
        MinaConfigInfo.getInStance().removeDisposable(tag);
        tcpBehaviorSubjectClient = null;
    }
}
